package com.swsc.ai.utils;

import com.swsc.ai.enums.CommonResultCode;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.springframework.stereotype.Component;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.FileOutputStream;
import java.io.InputStream;
import java.io.File;
import java.io.OutputStream;
import java.net.URI;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;

/**
 * @describe: ftp工具类，用于封装FTP操作的方法
 * @author: 容若
 * @created: 2024-06-20 11:20
 */
@Component
@Slf4j
public class FtpUtil {
    /**
     * 连接超时时间设置 5min
     */
    private static final int timeout = 1000 * 60 * 5;

    /**
     * ftp客户端创建
     */
    private static FTPClient ftpClient = new FTPClient();

    /**
     * 连接ftp服务
     *
     * @param host
     * @param port
     * @param username
     * @param password
     * @throws IOException
     */
    public void connect(String host, int port, String username, String password) throws IOException {
        ftpClient.connect(host, port);
        int replyCode = ftpClient.getReplyCode();
        if (!FTPReply.isPositiveCompletion(replyCode)) {
            ftpClient.disconnect();
            throw new IOException("FTP服务器拒绝连接!");
        }
        boolean success = ftpClient.login(username, password);
        if (!success) {
            ftpClient.logout();
            throw new IOException("FTP登录失败！");
        }
    }

    /**
     * 断开ftp连接
     */
    public void disconnect() {
        if (ftpClient.isConnected()) {
            try {
                ftpClient.logout();
                ftpClient.disconnect();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 判断ftp服务器文件是否存在
     *
     * @param path
     * @return
     */
    public boolean existFile(String path) {
        boolean flag = false;
        try {
            FTPFile[] ftpFileArr = ftpClient.listFiles(path);
            if (ftpFileArr.length > 0) {
                flag = true;
            }
        } catch (IOException e) {
            return false;
        }
        return flag;
    }

    /**
     * 列出目录下文件名
     *
     * @param directory
     * @return
     * @throws IOException
     */
    public List<String> listFiles(String directory) throws IOException {
        List<String> fileNames = new ArrayList<>();
        ftpClient.changeWorkingDirectory(directory);
        String[] names = ftpClient.listNames();
        for (String name : names) {
            fileNames.add(name);
        }
        return fileNames;
    }

    /**
     * 上传
     *
     * @param localFilePath  本地文件路径
     * @param remoteFilePath 上传到服务器路径
     * @throws IOException
     */
    public void uploadFile(String localFilePath, String remoteFilePath) throws IOException {
        InputStream inputStream = new FileInputStream(localFilePath);
//        InputStream inputStream1 = Files.newInputStream(Paths.get(localFilePath));
        boolean done = ftpClient.storeFile(remoteFilePath, inputStream);
        inputStream.close();
        if (!done) {
            throw new IOException("File upload failed");
        }
    }

    /**
     * 下载到本地
     *
     * @param remoteFilePath ftp服务器路径
     * @param localFilePath 本地存储路径
     * @throws IOException
     */
    public String downloadFileToLocal(String remoteFilePath, String localFilePath) throws IOException {
        ftpClient.setConnectTimeout(timeout);
        ftpClient.setControlEncoding("UTF-8"); //
        ftpClient.setFileType(FTPClient.BINARY_FILE_TYPE); // 二进制
        /**
         * 设置ftp被动模式
         */
        ftpClient.enterLocalPassiveMode();
        InputStream inputStream = ftpClient.retrieveFileStream(remoteFilePath);
        /**
         * 写入本地
         */
        FileOutputStream os = new FileOutputStream(localFilePath);
        byte[] buffer = new byte[1024];
        int bytesRead;
        while ((bytesRead = inputStream.read(buffer)) != -1) {
            os.write(buffer, 0, bytesRead);
        }
        inputStream.close();
        os.close();
        log.info("写文件到本地成功，路径：{}", localFilePath);
        return "写文件到本地成功，路径：" + localFilePath;
    }

    /**
     * 下载到HDFS
     * @param remoteFilePath  ftp服务器路径
     * @param hdfsFilePath hdfs存储文件路径
     * @param hdfsPath hdfs服务器路径
     * @param hadoopUser hadoop用户名
     * @throws IOException
     */
    public String downloadFileToHDFS(String remoteFilePath, String hdfsFilePath, String hdfsPath, String hadoopUser) throws IOException {
        ftpClient.setConnectTimeout(timeout);
        ftpClient.setControlEncoding("UTF-8"); //
        ftpClient.setFileType(FTPClient.BINARY_FILE_TYPE); // 二进制
        /**
         * 设置ftp被动模式
         */
        ftpClient.enterLocalPassiveMode();
        InputStream inputStream = ftpClient.retrieveFileStream(remoteFilePath);
        /**
         * 配置HDFS
         */
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", hdfsPath);
        try {
            FileSystem fileSystem = FileSystem.get(new URI(hdfsPath), conf,  hadoopUser);
            /**
             * 打开一个输出流到HDFS文件系统中的文件
             */
            org.apache.hadoop.fs.FSDataOutputStream outputStream = fileSystem.create(new Path(hdfsFilePath));
            /**
             * 使用Hadoop的IOUtils来复制数据
             */
            IOUtils.copyBytes(inputStream, outputStream, conf, true);
        } catch (Exception e) {
            log.info("写文件到HDFS异常：{}", e.toString());
            return CommonResultCode.ERROR.getMsg();
        }
        log.info("写文件到HDFS成功，路径为：{}", hdfsFilePath);
        return "写文件到HDFS成功，路径为：" + hdfsFilePath;
    }

    /**
     * Description: 从FTP服务器下载文件
     *
     * @param host       FTP服务器hostname
     * @param port       FTP服务器端口
     * @param username   FTP登录账号
     * @param password   FTP登录密码
     * @param remotePath FTP服务器上的相对路径
     * @param fileName   要下载的文件名
     * @param localPath  下载后保存到本地的路径
     * @return
     */
    public static boolean downloadFile(String host, int port, String username,
                                       String password, String remotePath,
                                       String fileName, String localPath) {
        boolean result = false;
        FTPClient ftp = new FTPClient();
        try {
            int reply;
            ftp.connect(host, port);
            /**
             * 如果采用默认端口，可以使用ftp.connect(host)的方式直接连接FTP服务器
             */
            ftp.login(username, password);// 登录
            reply = ftp.getReplyCode();
            if (!FTPReply.isPositiveCompletion(reply)) {
                ftp.disconnect();
                return result;
            }
            ftp.changeWorkingDirectory(remotePath);// 转移到FTP服务器目录
            FTPFile[] fs = ftp.listFiles();
            for (FTPFile ff : fs) {
                if (ff.getName().equals(fileName)) {
                    File localFile = new File(localPath + "/" + ff.getName());
                    OutputStream is = new FileOutputStream(localFile);
                    ftp.retrieveFile(ff.getName(), is);
                    is.close();
                }
            }
            ftp.logout();
            result = true;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (ftp.isConnected()) {
                try {
                    ftp.disconnect();
                } catch (IOException ioe) {
                }
            }
        }
        return result;
    }
}
